package com.icbc.dccsh.bolt;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintWriter;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class FlatFileBolt extends BaseBasicBolt {
	private PrintWriter _output;
	private String _base;
	private int _counter = 0;

	public FlatFileBolt(String base) {
		this._base = base;
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		this._counter = 0;
		String outputName = _base + context.getThisComponentId() + "-" + context.getThisTaskId();
		try {
			_output = new PrintWriter(new FileOutputStream(outputName));
		} catch (FileNotFoundException e) {
			throw new RuntimeException("Can't find output file for FileBolt: " + outputName, e);
		}
	}

	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		this._counter++;
		String separator = "";
		for (Object v : input.getValues()) {
			_output.print(separator);
			_output.print(v);
			separator = "\t";
		}
		_output.println();
		if (_counter % 100 == 0){
			_output.flush();
		}
	}
	@Override
	  public void cleanup() {
		_output.close();
	  }

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}
}
